#ifndef __M_SERVER_H__
#define __M_SERVER_H__
#include <iostream>
#include <string>
#include <vector>
#include <unordered_map>
#include <functional>
#include <memory>
#include <thread>
#include <mutex>
#include <condition_variable>
#include <cassert>
#include <cstdlib>
#include <cstring>
#include <ctime>
#include <pthread.h>
#include <unistd.h>
#include <signal.h>
#include <sys/types.h>
#include <arpa/inet.h>
#include <netinet/in.h>
#include <sys/socket.h>
#include <sys/epoll.h>
#include <sys/timerfd.h>
#include <sys/eventfd.h>
#include <fcntl.h>

/* 1.我们的这个高性能主从Reactor服务器组件写完之后，如果要提供给其他人使用，则有两种方式，一种是提供头文件.h + 库文件.o，别人在使用的时候
包含.h文件，然后在链接的时候链接上动态库即可，动态库里面其实就是我们所实现的各种方法编译之后的二进制代码，也就是方法的集合。
   2.而另一种也很常见的方式就是把方法的实现和方法的声明合在一起，搞成一个.hpp文件，对方直接包含这个.hpp文件就可以使用，而无需链接额外的库，
库其实就是.o文件集合
   3.我们的组件提供方式采用第二种，这样比较简单，因为第一种还得先编译，然后再打包.o文件，而不同的打包工具和打包方式还会划分出
静态库和动态库，感觉有点麻烦，使用者使用的时候还得链接库，不是很方便，所以我们采用第二种方式来提供给使用者 */

#define NOR 0
#define DBG 1
#define ERR 2
#define LOG_LEVEL DBG

#define LOG(level, format, ...)                                                                                        \
    do                                                                                                                 \
    {                                                                                                                  \
        if (level >= LOG_LEVEL)                                                                                        \
        {                                                                                                              \
            time_t t = time(NULL);                                                                                     \
            struct tm *local = localtime(&t);                                                                          \
            char tmp[32];                                                                                              \
            strftime(tmp, sizeof(tmp) - 1, "%H:%M:%S", local);                                                         \
            pthread_t tid = pthread_self();                                                                            \
            fprintf(stdout, "[%lu %s %s:%d]" format "\n", (unsigned long)tid, tmp, __FILE__, __LINE__, ##__VA_ARGS__); \
        }                                                                                                              \
    } while (0)

#define NLOG(format, ...) LOG(NOR, format, ##__VA_ARGS__)
#define DLOG(format, ...) LOG(DBG, format, ##__VA_ARGS__)
#define ELOG(format, ...) LOG(ERR, format, ##__VA_ARGS__)

#define BUFFER_DEFAULT_SIZE 1024
class Buffer
{
private:
    // 使用char而不是 string 来保存数据，因为 string 遇到'\0'就会结束，而通信数据中可能会包含'\0'
    std::vector<char> _buffer;
    uint64_t _read_idx;  // 读偏移
    uint64_t _write_idx; // 写偏移
public:
    Buffer()
        : _buffer(BUFFER_DEFAULT_SIZE), _read_idx(0), _write_idx(0)
    {
    }

    char *WriteAddress() { return &_buffer[_write_idx]; }

    char *ReadAddress() { return &_buffer[_read_idx]; }

    const char *ReadAddress() const { return &_buffer[_read_idx]; }

    // 获取_write_idx后面空闲空间大小
    uint64_t BackRemain() { return _buffer.size() - _write_idx; }

    // 获取_read_idx前面空闲空间大小
    uint64_t FrontRemain() { return _read_idx; }

    // 开区间 减去 闭区间
    uint64_t ReadableSize() const { return _write_idx - _read_idx; }

    // 读取数据之后，将_read_idx向后移动至正确位置
    void MoveReadIdx(uint64_t len)
    {
        assert(len <= ReadableSize());
        _read_idx += len;
    }

    // 写入数据之后，将_write_idx向后移动至正确位置
    void MoveWriteIdx(uint64_t len)
    {
        assert(len <= BackRemain());
        _write_idx += len;
    }

    // 确保可写空间是足够的
    void EnsureWriteSpace(uint64_t size)
    {
        if (BackRemain() >= size)
        {
            // 1. _write_idx后面空间足够，啥也不用干
        }
        else if (FrontRemain() + BackRemain() >= size)
        {
            // 2. 整体空间足够，移动数据至开头
            uint64_t rsz = ReadableSize();
            std::copy(ReadAddress(), ReadAddress() + rsz, &_buffer[0]); // copy第二个参数时开区间

            _read_idx = 0;
            _write_idx = rsz; // _write_idx是开区间
        }
        else
        {
            // 3. 整体空间不够，不移动数据，直接扩容
            _buffer.resize(_write_idx + size);
        }
        return;
    }

    // 写入数据
    void Write(const void *data, uint64_t size)
    {
        // 1. 确保有足够的可写空间
        EnsureWriteSpace(size);

        // 2. 写入数据(拷贝)
        std::copy((const char *)data, (const char *)data + size, WriteAddress());
    }

    // 写入数据 + 挪动写偏移量
    void WriteAndMoveIdx(const void *data, uint64_t size)
    {
        Write(data, size);
        MoveWriteIdx(size);
    }

    void Write(const std::string &data)
    {
        Write(data.c_str(), data.size());
    }

    void WriteAndMoveIdx(const std::string &data)
    {
        Write(data);
        MoveWriteIdx(data.size());
    }

    void Write(const Buffer &data)
    {
        Write(data.ReadAddress(), data.ReadableSize());
    }

    void WriteAndMoveIdx(const Buffer &data)
    {
        Write(data);
        MoveWriteIdx(data.ReadableSize());
    }

    // 读取数据
    void Read(void *buf, uint64_t size)
    {
        assert(size <= ReadableSize());
        std::copy(ReadAddress(), ReadAddress() + size, (char *)buf);
    }

    // 读取数据 + 挪动写偏移量
    void ReadAndMoveIdx(void *buf, uint64_t size)
    {
        Read(buf, size);
        MoveReadIdx(size);
    }

    std::string Read(uint64_t size)
    {
        assert(size <= ReadableSize());
        std::string str;
        str.resize(size);
        Read(&str[0], size);
        return str;
    }

    std::string ReadAndMoveIdx(uint64_t size)
    {
        assert(size <= ReadableSize());
        std::string str = Read(size);
        MoveReadIdx(size);
        return str;
    }

    // 清空缓冲区
    void Clear()
    {
        // 只需要将偏移量归0即可！
        _read_idx = 0;
        _write_idx = 0;
        return;
    }

    // 由于后面会用到HTTP协议，而在HTTP中经典的操作就是读取一行的数据，
    // 因为请求行和请求报头以及响应行和响应报头都是以r\n作为分隔符的，都是一行一行的数据
    // 所以我们的缓冲区这里也对外提供一个接口，用于每次读取一行的数据
    char *FindCRLF()
    {
        void *ret = memchr(ReadAddress(), '\n', ReadableSize());
        return (char *)ret;
    }

    // 从缓冲区中取出一行HTTP请求数据
    std::string GetLine()
    {
        char *pos = FindCRLF();
        if (pos == NULL)
            return "";
        // + 1是为了把'\n'也给读进去
        return Read(pos - ReadAddress() + 1);
    }

    std::string GetLineAndMoveIdx()
    {
        std::string str = GetLine();
        MoveReadIdx(str.size());
        return str;
    }
};

// 创建套接字
// 绑定地址信息
// 开始监听
// 向服务器发起连接
// 获取新连接
// 接受数据
// 发送数据
// 关闭套接字
// 创建一个监听链接
// 创建一个客户端连接
#define MAX_LISTEN 1024
class Socket
{
private:
    int _sockfd;

public:
    Socket()
        : _sockfd(-1)
    {
    }

    Socket(int sockfd)
        : _sockfd(sockfd)
    {
    }

    ~Socket() { Close(); }

    // 给上层一个获取fd的接口
    int Fd() { return _sockfd; }

    // 1. 创建套接字
    bool Create()
    {
        // int socket (int domain,int type,int protocol);
        _sockfd = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
        if (_sockfd < 0)
        {
            ELOG("create socket failed!!!");
            return false;
        }
        return true;
    }

    // 2. 绑定地址信息
    bool Bind(const std::string &ip, uint16_t port)
    {
        struct sockaddr_in addr;
        addr.sin_family = AF_INET;
        addr.sin_port = htons(port);
        addr.sin_addr.s_addr = inet_addr(ip.c_str());
        socklen_t len = sizeof(struct sockaddr_in);
        // int bind(int sockfd,struct sockaddr * addr,socklen_t len);
        int ret = bind(_sockfd, (const struct sockaddr *)&addr, len);
        if (ret < 0)
        {
            ELOG("bind address failed!!!");
            return false;
        }
        return true;
    }

    // 3. 监听
    bool Listen(int backlog = MAX_LISTEN)
    {
        int ret = listen(_sockfd, backlog);
        if (ret < 0)
        {
            ELOG("socket listen failed!!!");
            return false;
        }
        return true;
    }

    // 4. 向服务器发起连接
    bool Connect(const std::string &ip, uint16_t port)
    {
        struct sockaddr_in addr;
        addr.sin_family = AF_INET;
        addr.sin_port = htons(port);
        addr.sin_addr.s_addr = inet_addr(ip.c_str());
        socklen_t len = sizeof(struct sockaddr_in);

        // int connect(int sockfd，const struct sockaddr *addr, socklen_t addrlen);
        int ret = connect(_sockfd, (const struct sockaddr *)&addr, len);
        if (ret < 0)
        {
            ELOG("connect address failed!!!");
            return false;
        }
        return true;
    }

    // 5. 底层有连接就绪时，获取新连接
    int Accpet()
    {
        // int accept(int sockfd, struct sockaddr * addr, socklen_ _t *addrlen);
        // 后两个参数是输出型参数，用于给服务器返回客户端的地址信息
        // 目前服务器并不关心客户端的地址信息，所以我们直接传NULL
        int newfd = accept(_sockfd, NULL, NULL);
        if (newfd < 0)
        {
            ELOG("socket accpet failed!!!");
            return -1;
        }
        return newfd;
    }

    // 6. 读取数据
    ssize_t Recv(void *buf, size_t len, int flag = 0) // 标志位可以将接收数据设置为非阻塞操作，默认是0阻塞操作
    {
        // size_t 是无符号整数
        // ssize_t 是有符号整数
        ssize_t ret = recv(_sockfd, buf, len, flag);
        if (ret <= 0)
        {
            if (errno == EAGAIN || errno == EINTR)
            {
                // 非阻塞读取时才会出现EAGAIN这样的错误码，表示当前内核缓冲区没有数据了，希望你下一次再进行读取
                // EINTR表示被信号打断，转而处理信号函数去了，那这种也不算错误
                return 0;
            }
            else
            {
                // 这样才算是真的出错了
                ELOG("socket recv failed!!!");
                return -1;
            }
        }
        return ret;
    }

    ssize_t NonBlockRecv(void *buf, size_t len)
    {
        // MSG_DONTWAIT表示非阻塞接收数据
        return Recv(buf, len, MSG_DONTWAIT);
    }

    // 7. 发送数据
    ssize_t Send(const void *buf, size_t len, int flag = 0)
    {
        ssize_t ret = send(_sockfd, buf, len, flag);
        if (ret < 0)
        {
            if (errno == EAGAIN || errno == EINTR)
            {
                // 非阻塞读取时才会出现EAGAIN这样的错误码，表示当前内核缓冲区没有数据了，希望你下一次再进行读取
                // EINTR表示被信号打断，转而处理信号函数去了，那这种也不算错误
                return 0;
            }
            else
            {
                // 这样才算是真的出错了
                ELOG("socket send failed!!!");
                return -1;
            }
        }
        // 实际发送的数据长度，有可能实际发送的长度比想要发送的长度1en要小，因为缓冲区中可能没有1en那么多的数据
        return ret;
    }

    ssize_t NonBlockSend(void *buf, size_t len)
    {
        if (len <= 0)
            return 0;
        // 发送数据操作为非阻塞，如果socket内核接收缓冲区中没有多余空间，则非阻塞返回后，errno为EAGAIN
        return Send(buf, len, MSG_DONTWAIT);
    }

    // 8. 关闭套接字
    void Close()
    {
        if (_sockfd != 1)
        {
            close(_sockfd);
            _sockfd = -1;
        }
    }

    // 9. 创建服务端连接
    bool CreateServerConn(uint16_t port, bool block_flag = false, const std::string &ip = "0.0.0.0")
    {
        // 1. 创建套接字  2. 绑定地址信息    3. 设置套接字为非阻塞  4. 监听  5. 设置套接字地址复用
        if (Create() == false)
            return false;

        if (Bind(ip, port) == false)
            return false;

        if (block_flag)
            SetNonBlock();

        if (Listen() == false)
            return false;

        SetReuseAddress();
        return true;
    }

    // 10. 创建客户端连接
    bool CreateClientConn(uint16_t port, const std::string &ip)
    {
        // 1. 创建套接字  2. 绑定地址信息(不需要显示绑定，在第一次send的时候OS自动帮我们绑定，包括IP)  3. 发起连接
        if (Create() == false)
            return false;

        if (Connect(ip, port) == false)
            return false;
        return true;
    }

    // 11. 设置套接字地址复用
    void SetReuseAddress()
    {
        int opt = 1;
        setsockopt(_sockfd, SOL_SOCKET, SO_REUSEADDR, (void *)&opt, sizeof(int));

        opt = 1;
        setsockopt(_sockfd, SOL_SOCKET, SO_REUSEPORT, (void *)&opt, sizeof(int));
        // SO_REUSEPORT避免了应用层多线程或者进程监听同一ip: port时的"惊群效应"
    }

    // 12. 设置文件描述符为非阻塞操作(防止recv不断读取数据，使得内核缓冲区中没有数据时，执行流阻塞在recv读取这块儿)
    void SetNonBlock()
    {
        int flag = fcntl(_sockfd, F_GETFL, 0);
        fcntl(_sockfd, F_SETFL, flag | O_NONBLOCK);
    }
};

class Poller;
class EventLoop;

// 对一个fd监控事件的管理(注册，改变，移除)
// 分派事件，当fd的事件就绪之后，调用用户设定的回调函数
class Channel
{
    using EventCallback = std::function<void()>;

private:
    int _fd;
    EventLoop *_loop;
    uint32_t _events;  // 当前需要监控的事件
    uint32_t _revents; // 当前连接触发的事件(已经就绪的事件集合)

    EventCallback _read_cb;     // socket上可读事件就绪时的回调
    EventCallback _write_cb;    // socket上可写事件就绪时的回调
    EventCallback _error_cb;    // socket上错误事件就绪时的回调
    EventCallback _close_cb;    // socket上连接关闭事件就绪时的回调
    EventCallback _anyevent_cb; // socket上任意事件就绪时的回调(刷新活跃度)

public:
    Channel(int fd, EventLoop *loop)
        : _fd(fd), _loop(loop), _events(0)
    {
    }

    int Fd() { return _fd; }

    uint32_t GetEvents() { return _events; }

    // 设置就绪的事件
    void SetRevents(uint32_t revents) { _revents = revents; }

    void SetReadCallback(const EventCallback &cb)
    {
        _read_cb = cb;
    }
    void SetWriteCallback(const EventCallback &cb)
    {
        _write_cb = cb;
    }
    void SetErrorCallback(const EventCallback &cb)
    {
        _error_cb = cb;
    }
    void SetCloseCallback(const EventCallback &cb)
    {
        _close_cb = cb;
    }
    void SetAnyEventCallback(const EventCallback &cb)
    {
        _anyevent_cb = cb;
    }

    // 当前是否可读
    bool IsReadable() { return (_events & EPOLLIN); }

    // 当前是否可写
    bool IsWriteable() { return (_events & EPOLLOUT); }

    // 启动对读事件的关心，后面会添加到EventLoop的事件监控！
    void CareReadEvent()
    {
        _events |= EPOLLIN;
        UpdateEvents();
    }

    // 启动对写事件的关心，后面会添加到EventLoop的事件监控！
    void CareWriteEvent()
    {
        _events |= EPOLLOUT;
        UpdateEvents();
    }

    // 取消对读事件的关心，后面会修改到EventLoop的事件监控！
    void CancelReadEvent()
    {
        _events &= ~EPOLLIN;
        UpdateEvents();
    }

    // 取消对写事件的关心，后面会修改到EventLoop的事件监控！
    void CancelWriteEvent()
    {
        _events &= ~EPOLLOUT;
        UpdateEvents();
    }

    // 取消对所有事件的关心，后面会修改到EventLoop的事件监控！
    void CancelAllEvent()
    {
        _events = 0;
        UpdateEvents();
    }

    // 后面会调用EventLoop接口刷新活跃度
    void UpdateEvents();

    // 后面会调用EventLoop接口移除监控
    void RemoveEvents();

    // eventloop不关心如何处理就绪的事件，Channel自己负责，EventLoop只负责调用channel里的HandleReadyEvent()
    void HandlerReadyEvents()
    {
        // 后面实现时，读取出错是不会释放连接的，即使对方挂断连接，我们读取出错，我们也要看outbuffer中是否还有待send的data
        // 如果有，我们还要将待send的data进行发送，此时给已经断开连接的对方发送数据自然会出错，这个时候我们才会去断开连接
        // (所以不会出现由于重复释放连接，而导致segmentation fault)
        if ((_revents & EPOLLIN) || (_revents & EPOLLPRI) || (_revents & EPOLLRDHUP))
        {
            if (_read_cb)
                _read_cb();
        }

        // 有可能当前触发了可写事件，但写入数据不成功，对方已经挂断了连接，此时如果继续向下进行处理则会导致重复释放，所以是if-else
        if (_revents & EPOLLOUT)
        {
            if (_write_cb)
                _write_cb();
        }
        else if (_revents & EPOLLERR)
        {
            if (_error_cb)
                _error_cb();
        }
        else if (_revents & EPOLLHUP)
        {
            if (_close_cb)
                _close_cb();
        }

        // 在刷新活跃度之前，即使连接已经断开，但上面的接口都不能真正把连接释放
        // 因为下面还会进行连接活跃度的刷新，所以在上面的回调函数中必须将实际释放连接的操作压入到任务池
        // 在就绪事件全部处理完毕之后，再下一步执行任务池中的任务时，才能进行实际连接的释放
        if (_anyevent_cb)
            _anyevent_cb(); // fd上任意事件触发，刷新连接的活跃度
    }
};

// 就绪数组一次能够容纳的最大事件数量
#define MAX_EPOLLEVENTS 1024

// epoll_create参数，此参数现已被忽略，可以随便设，但要求大于0
#define MAX_SIZE 1024

// Poller模块是对epoll进⾏封装的⼀个模块，主要实现epoll的IO事件添加，修改，移除，获取活跃连接功能。
class Poller
{
private:
    int _epfd; // epoll的操作句柄(epoll模型)
    // 就绪队列
    struct epoll_event _evs[MAX_EPOLLEVENTS];
    // 使用hash表管理描述符与描述符对应的事件管理Channel对象
    // 当描述符就绪了，通过描述符在hash表中找到对应的Channel(得到了Channel才知道什么事件如何处理)
    // 当描述符就绪了，返回就绪描述符对应的Channel
    std::unordered_map<int, Channel *> _chans;

private:
    // 1. 判断一个Channel是否已经添加到了事件监控
    bool HasChannel(Channel *channel)
    {
        auto iter = _chans.find(channel->Fd());
        if (iter == _chans.end())
            return false;
        return true;
    }

    // 2. 针对epoll直接操作(添加，修改，移除)
    void Update(Channel *channel, int op)
    {
        int fd = channel->Fd();
        struct epoll_event ev;
        ev.data.fd = fd;
        ev.events = channel->GetEvents();

        int ret = epoll_ctl(_epfd, op, fd, &ev);
        if (ret < 0)
            ELOG("epoll ctl failed!!!");
    }

public:
    Poller()
    {
        _epfd = epoll_create(MAX_SIZE); // 参数随便给值就行，大于0就行
        if (_epfd < 0)
        {
            ELOG("create epoll error!!!");
            abort();
        }
    }

    // Channel中既包含了操作的套接字描述符，又包含了描述符上关心的事件有哪些，还包含了事件就绪时的回调函数
    void UpdateEvents(Channel *channel)
    {
        bool ret = HasChannel(channel);
        if (ret == false)
        {
            // 不存在则将fd-channel*添加至哈希表中
            _chans.insert(std::pair<int, Channel *>(channel->Fd(), channel));
            Update(channel, EPOLL_CTL_ADD);
            return;
        }
        // 存在则modify红黑树节点中fd所关心的事件内容
        Update(channel, EPOLL_CTL_MOD);
        return;
    }

    void RemoveEvents(Channel *channel)
    {
        // 先移除事件的关心，再释放_chans的对应信息
        Update(channel, EPOLL_CTL_DEL);

        auto iter = _chans.find(channel->Fd());
        if (iter != _chans.end())
            _chans.erase(iter);
    }

    // 开始监控，返回活跃的channel数组，表示某些fd上的某些关心的事件已经ready了
    void Monitor(std::vector<Channel *> *active)
    {
        // -1表示阻塞监控，如果没有事件就绪，服务器进程执行流就会一直阻塞在这里
        int nfds = epoll_wait(_epfd, _evs, MAX_EPOLLEVENTS, -1);
        if (nfds < 0)
        {
            // 被信号中断
            if (errno == EINTR)
                return;
            else
            {
                ELOG("epoll wait error: %s", strerror(errno));
                abort();
            }
        }

        for (int i = 0; i < nfds; i++)
        {
            auto iter = _chans.find(_evs[i].data.fd);
            assert(iter != _chans.end());

            // 设置channel事件管理器中已经ready的事件
            iter->second->SetRevents(_evs[i].events);
            active->push_back(iter->second);
        }
    }
};

using TaskFunc = std::function<void()>;
using ReleaseFunc = std::function<void()>;

class Timer
{
private:
    uint64_t _id;         // 定时器任务id
    uint32_t _timeout;    // 定时器超时时间
    bool _canceled;       // 定时器是否被取消
    TaskFunc _task_cb;    // 定时器超时时应该执行的任务函数
    ReleaseFunc _release; // 定时器超时时除了要执行任务函数之外，还需要把哈希表中的weak_ptr销毁

public:
    Timer(uint64_t id, uint32_t timeout, TaskFunc task_cb)
        : _id(id), _timeout(timeout), _task_cb(task_cb), _canceled(false)
    {
    }

    ~Timer()
    {
        // 定时器如果未被取消，则执行定时任务
        if (_canceled == false)
            _task_cb();

        // 释放时间轮中保存的weak_ptr定时器对象信息
        _release();
    }

    // 设置回调函数执行释放时间轮中保存的weak_ptr定时器对象信息
    void setDestructor(ReleaseFunc cb) { _release = cb; }

    // 获取超时时间
    int timeOut() { return _timeout; }

    // 取消定时器任务
    void cancel() { _canceled = true; }
};

class TimerQueue
{
private:
    int _tick;     // 秒针
    int _capacity; // 表盘最大刻度 - 最大延迟时间

    EventLoop *_loop;

    // 时间轮需要管理多个Timer对象，这里使用哈希表进行管理(id - weak_ptr<Timer>) 使用weak_ptr是因为不会增加引用计数
    std::unordered_map<uint64_t, std::weak_ptr<Timer>> _timers;

    // 保存多个Timer对象的shared_ptr，实现秒级时间轮
    std::vector<std::vector<std::shared_ptr<Timer>>> _wheel;

    // 将linux系统的定时器timerfd融入到TimerQueue中，实现出一个秒级定时任务的时间轮出来
    int _timer_fd;
    std::unique_ptr<Channel> _timer_chan; // timerfd也是一个fd，以后也有事件需要监控

private:
    // 释放时间轮中保存的weak_ptr定时器对象信息(我们让每个定时都绑定这个方法，在定时器销毁时去除哈希表的映射关系)
    void ReleaseWeakPtr(uint64_t tid)
    {
        auto iter = _timers.find(tid);
        if (iter != _timers.end())
            _timers.erase(iter);
    }

    // 激活秒级定时器
    static int ActivateTimer()
    {
        // CLOCK_MONOTONIC-从开机到现在的时间是⼀种相对时间；
        // flags: 0-默认阻塞属性
        int timerfd = timerfd_create(CLOCK_MONOTONIC, 0);
        if (timerfd < 0)
        {
            ELOG("timerfd create error!!!");
            abort();
        }

        struct itimerspec itime;
        // 第一次超时时间设为1s
        itime.it_value.tv_sec = 1;
        itime.it_value.tv_nsec = 0;

        // 第一次超时后，每隔1s算一次超时时间
        itime.it_interval.tv_sec = 1;
        itime.it_interval.tv_nsec = 0;

        // 启动定时器，让1inux系统每隔1s向定时器文件描述符中写入1次超时次数的数据
        // int timerfd_settime(int fd, int flags, struct itimerspec *new, struct itimerspec *old);
        // fd: timerfd_create返回的⽂件描述符
        // flags: 0-相对时间， 1-绝对时间；默认设置为0即可.
        // new： ⽤于设置定时器的新超时时间
        // old： ⽤于接收原来的超时时间
        timerfd_settime(timerfd, 0, &itime, NULL);
        return timerfd;
    }

    int ReadTimerfd()
    {
        u_int64_t timeout;
        int ret = read(_timer_fd, &timeout, sizeof timeout);
        if (ret < 0)
        {
            ELOG("read timerfd failed!!!");
            abort();
        }
        return timeout;
    }

    void RunTimer()
    {
        _tick = (_tick + 1) % _capacity;
        // 析构Timer定时器对象
        _wheel[_tick].clear();
    }

    // 当timerfd的可读事件就绪时，就会执行该回调函数，读出超时的次数，析构tick秒针沿途经历过的各个Timer定时器对象
    void HandlerTimer()
    {
        // count表示经历了多少s，因为我们是每隔1s超时一次
        int count = ReadTimerfd();
        for (int i = 0; i < count; i++)
            RunTimer();
    }

private:
    void AddTimerPriv(uint64_t tid, int timeout, TaskFunc task_cb)
    {
        // 1. 构造一个shared_ptr管理的定时器Timer对象，同时设置一下Timer对象的成员变量，让Timer对象析构时，释放掉时间轮保存的Timer信息
        std::shared_ptr<Timer> sp(new Timer(tid, timeout, task_cb));
        sp->setDestructor(std::bind(&TimerQueue::ReleaseWeakPtr, this, tid));

        // 2. 将该任务对象添加到时间轮当中
        int pos = (_tick + timeout) % _capacity;
        _wheel[pos].push_back(sp);

        // 3. 将定时器的weak_ptr添加至哈希表中
        _timers[tid] = std::weak_ptr<Timer>(sp);
    }

    void FlushTimerPriv(uint64_t tid)
    {
        // 1. 通过保存的weak_ptr定时器对象，构造一个shared_ptr出来，将shared_ptr插入到wheel时间轮当中
        auto iter = _timers.find(tid);
        if (iter == _timers.end())
            return;

        std::shared_ptr<Timer> sp = iter->second.lock();

        // 2. 获取定时任务的延迟时间
        int delay = sp->timeOut();

        // 3. 确定添加sp的位置
        int pos = (_tick + delay) % _capacity; // tick走的下标的范围是[0, 59]
        _wheel[pos].push_back(sp);             // 刷新定时任务
    }

    void CancelTimerPriv(uint64_t tid)
    {
        // 1. 通过保存的weak_ptr定时器对象，构造一个shared_ptr出来，将shared_ptr插入到wheel时间轮当中
        auto iter = _timers.find(tid);
        if (iter == _timers.end())
            return;

        std::shared_ptr<Timer> sp = iter->second.lock();

        // 2. 取消定时器任务
        if (sp)
            sp->cancel();
    }

public:
    TimerQueue(EventLoop *loop)
        : _tick(0), _capacity(60), _wheel(60), _loop(loop), _timer_fd(ActivateTimer()), _timer_chan(new Channel(_timer_fd, loop))
    {
        _timer_chan->SetReadCallback(std::bind(&TimerQueue::HandlerTimer, this));
        _timer_chan->CareReadEvent();
    }

    void AddTimer(uint64_t tid, int timeout, TaskFunc task_cb);
    void FlushTimer(uint64_t tid);
    void CancelTimer(uint64_t tid);

    bool HasTimer(uint64_t tid)
    {
        auto iter = _timers.find(tid);
        if (iter == _timers.end())
            return false;
        return true;
    }
};

// Eventloop事件循环类
// 描述符事件监控以及事件处理类
class EventLoop
{
    using Func = std::function<void()>;

private:
    std::thread::id _thread_id;             // EventLoop关联的线程ID
    Poller _poller;                         // 描述符IO事件监控模块
    int _event_fd;                          // eventfd -- 唤醒IO事件监控可能导致的阻塞
    std::unique_ptr<Channel> _evfd_channel; // 管理_event_fd的监控事件
    std::mutex _mutex;                      // 实现任务队列的线程安全
    std::vector<Func> _tasks;               // 任务队列
    TimerQueue _timer_queue;                // 定时器模块

private:
    // 执行任务队列中的所有任务
    void ExecTasks()
    {
        // 如果给队列加锁，执行队列中的任务就需要加锁，而向队列中push任务就需要解锁，这样的效率很低。
        // 所以我们直接做一个vector级别的置换，将任务队列中的任务转移到另一个vector中，这样就可以保证任务队列的线程安全。
        std::vector<Func> tmp;
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _tasks.swap(tmp);
        }

        for (auto &f : tmp)
            f();
    }

    // 因为有可能因为等待描述符IO事件就绪，导致执行流流程阻塞
    // 这时候任务队列中的任务将得不到执行，因此需要使用 eventfd 来进行事件通知，唤醒事件监控的阻塞。
    // 创建event_fd
    static int CreateEventFd()
    {
        int init = 0;                                         // eventfd计数器的初始值
        int evfd = eventfd(init, EFD_CLOEXEC | EFD_NONBLOCK); // 对eventfd的操作都是非阻塞的
        if (evfd < 0)
        {
            ELOG("create eventfd failed!!!");
            abort();
        }
        return evfd;
    }

    // eventfd的可读事件回调
    void ReadEventFd()
    {
        // 读取eventfd中的数据，使其计数器重新变为0(8字节)
        uint64_t res = 0;
        ssize_t n = read(_event_fd, &res, sizeof res);
        if (n <= 0)
        {
            // 信号打断 + efd对应的内核接收缓冲区中没有数据
            if (errno == EINTR || errno == EAGAIN)
                return;
            ELOG("read eventfd failed!!!");
            abort();
        }
    }

    // 唤醒可能因为没有IO事件监控而导致的阻塞
    void WakeupEpollwait()
    {
        uint64_t val = 1; // 必须是8字节的数据
        int ret = write(_event_fd, &val, sizeof val);
        if (ret <= 0)
        {
            if (errno == EINTR)
                return; // 信号打断
            ELOG("write eventfd failed!!!");
            abort();
        }
    }

public:
    EventLoop()
        : _thread_id(std::this_thread::get_id()), _event_fd(CreateEventFd()), _evfd_channel(new Channel(_event_fd, this)), _timer_queue(this)
    {
        _evfd_channel->SetReadCallback(std::bind(&EventLoop::ReadEventFd, this));
        _evfd_channel->CareReadEvent();
        // DLOG("EventLoop初始化完毕, 对应线程ID是: %lu", _thread_id);
    }

    // 判断当前线程是否是EventLoop对应的线程
    bool IsLoopThread() { return _thread_id == std::this_thread::get_id(); }

    // 确保当前线程就是EventLoop对应的线程
    void AssertInLoop() { assert(_thread_id == std::this_thread::get_id()); }

    // 添加/修改描述符的监控事件
    void UpdateEvent(Channel *chan) { _poller.UpdateEvents(chan); }

    // 移除描述符的事件监控
    void RemoveEvent(Channel *chan) { _poller.RemoveEvents(chan); }

    // 添加定时任务
    void AddTimer(uint64_t id, int timout, const TaskFunc &cb) { _timer_queue.AddTimer(id, timout, cb); }

    // 刷新/延迟定时任务
    void FlushTimer(uint64_t id) { _timer_queue.FlushTimer(id); }

    // 取消定时任务
    void CancelTimer(uint64_t id) { _timer_queue.CancelTimer(id); }

    // 判断某个定时任务是否存在
    bool HasTimer(uint64_t id) { return _timer_queue.HasTimer(id); }

    // 判断将要执行的任务是否处于当前线程中，是则直接执行，不是则将其添加到任务队列中
    void RunInLoop(const Func &cb)
    {
        if (IsLoopThread())
        {
            cb();
        }
        else
        {
            PushInTasks(cb);
        }
    }

    // 将任务添加到任务队列中
    void PushInTasks(const Func &cb)
    {
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _tasks.push_back(cb);
        }

        // 唤醒可能因为没有IO事件监控而导致的阻塞
        // 压入任务池后，立马唤醒epoll_wait，让LoopThread线程继续向下进行就绪事件处理+执行任务池中的任务
        WakeupEpollwait();
    }

    // 三步走 --- 事件监控 -> 就绪事件处理 -> 执行任务
    void RunLoop()
    {
        while (1)
        {
            // 1. 事件监控
            // epoll_wait红黑树上所有的fd，将所有ready的fd所对应的channel输出到参数actives中
            std::vector<Channel *> actives;
            _poller.Monitor(&actives);

            // for (auto &chan : actives)
            // {
            //     DLOG("事件就绪, fd: %d", chan->Fd());
            // }

            // 2. 就绪事件处理
            for (auto &chan : actives)
                chan->HandlerReadyEvents();

            // 3. 执行任务
            ExecTasks();
        }
    }
};

// 我们上面在设计 EventLoop 模块时提到 EventLoop 模块与线程是一一对应的
// 并且由于 EventLoop 模块在构造时就会使用当前线程 id 来作为 EventLoop 对象所关联的线程的 _thread_id(std::this_thread::get_id())

// EventLoop 模块必须在线程内部实例化，即先为 EventLoop 对象创建一个线程
// 然后在该线程的入口函数中去实例化EventLoop对象，这样该线程就会与EventLoop对象相关联(实例化时该线程 id 被用于初始化EventLoop对象的_thread_id)

// 我们需要构建一个 LoopThread 模块，这个模块的功能是将 EventLoop 与 thread 整合到一起，向外部返回所实例化的 EventLoop 对象

// 需要注意一个问题就是线程间的时序问题
// 主线程可能会调用GetEventLoop接口获取EventLoop对象的指针，将这个指针关联到一个Connection连接上
// 但此时很有可能从线程还没有创建完毕EventLoop呢，那么这个时候GetEventLoop接口会返回一个NULL指针，这样肯定是不行的
// 所以在LoopThread这里需要借助 线程同步与互斥 的知识，即通过条件变量和互斥锁来控制线程间的运行

class LoopThread
{
private:
    std::thread _thread;
    // 互斥锁和条件变量用于实现获取_loop指针的同步关系，避免线程创建了但EventLoop还未实例化时获取_loop指针
    std::mutex _mutex;
    std::condition_variable _cond;
    // 注意这里定义的是一个指针，而不是直接定义对象，因为EventLoop对象的实例化必须放到线程的入口函数里面进行
    EventLoop *_loop;

private:
    // 线程创建时的函数入口，此时就可以去初始化EventLoop，并与之对应的线程关联起来了
    void ThreadEntry()
    {
        // 1.实例化EventLoop对象(我们在栈上创建对象，而不是堆，其目的就是将EventLoop对象的生命周期和LoopThread的生命周期进行绑定)
        // 这里创建局部对象即可，不必new，因为_loop.start是死循环运行的
        EventLoop loop; // 线程跑完这句代码时，就已经和EventLoop关联好了
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _loop = &loop;
            // 2.唤醒有可能阻塞在_cond上的线程
            _cond.notify_all();
        }

        // 3. 运行EventLoop模块 -- 获取并处理活跃连接，执行任务
        DLOG("创建子Reactor成功!!!");
        loop.RunLoop();
    }

public:
    LoopThread()
        : _loop(NULL), _thread(std::bind(&LoopThread::ThreadEntry, this)) // 创建线程，设定线程入口函数
    {
    }

    EventLoop *GetEventLoop()
    {
        EventLoop *loop = NULL;
        {
            std::unique_lock<std::mutex> lock(_mutex);
            // _loop为空就阻塞等待，wait阻塞等待时，就会释放互斥锁，直到_loop不为空则被唤醒
            _cond.wait(lock, [&]()
                       { return _loop != NULL; });
            loop = _loop;
            /* 实现线程间的同步操作:
            避免线程创建了，但此时EventLoop对象还没被实例化出来，此时如果直接调用GetEventLoop，则会返回一个NULL，
            所以我们实现LoopThread和mainThread的同步操作，mainThread调用GetEventLoop后，返回的一定是有效的EventLoop对象指针*/
            /* 线程同步的原理:
            通过锁的阻塞行为来在某一特定条件下阻塞一个线程，从而让另一个线程在完成某件功能后唤醒阻塞的线程，以此来实现线程间协调工作*/
        }
        return loop;
    }
};

// LoopThreadPool实际上就是用来创建LoopThread对象，只要该对象被创建则从Reactor线程就会被创建，同时也会关联一个EventLoop对象
// LoopThreadPool的主要功能其实就是创建从属Reactor线程
// 以及提供一个RR轮转分配EventLoop对象指针的接口，即GetNextEventLoop接口，即每一个服务器上新到来的连接都会被均匀分配到创建出来的一定个数的EventLoop上。

// 由于客户端连接有多个，而每一个客户端连接都对应一个 Connection 模块、EventLoop 模块以及 LoopThread 模块
// 因此我们需要针对 LoopThread 设计一个线程池 – LoopThreadPool，用于对所有的 LoopThread 进行管理及分配

/*
LoopThreadPool 模块所要完成的功能如下：
1. 线程数量可配置(0个或多个)
   在服务器中，由于主从 Reactor 模型是主线程只负责新连接获取，从属线程负责新连接的事件监控及处理
   因此当前的线程池中从属线程的数量有可能会为0，也就是实现单 Reactor 服务器，仅有一个线程，其即负责获取连接，也负责连接的处理。

2. 对所有的线程进行管理 – 管理0个或多个 LoopThread 对象

3. 提供线程分配的功能 – 当主线程获取了一个新连接时，将新连接挂到从属线程上进行事件监控及处理。
   假设有0个从属线程，则直接分配给主线程的 EventLoop 进行处理
   假设有多个从属线程，则采用 RR 轮转思想，进行线程的分配(将被选择线程的 EventLoop 对象获取到，然后设置给对应的 Connection 对象)
*/

class LoopThreadPool
{
private:
    int _thread_count;
    int _next_loop_idx; // 下一个EventLoop的索引(RR轮转分配)

    // 若从属线程数量为0，则事件的监控，就绪事件的处理，执行任务池中的任务等所有操作都在主线程的_baseloop内完成
    EventLoop *_baseloop;
    /*
    实际上这个baseloop是TcpServer创建的，因为主Reactor和从Reactor线程最终都一样

    他们最终其实都是死循环运行EventLoop模块的RunLoop接口，只不过主Reactor监控的其实是监听套接字上的可读事件什么时候能够就绪
    而从Reactor监控的是通信套接字上可能触发的可读事件，可写事件，错误事件，连接断开事件等，这就是两者的不同

    所以如果服务器组件是单Reactor单线程模式的话，则实际上监听套接字和通信套接字的事件监控与就绪事件处理等都会在baseloop中运行
    主线程则会一直死循环运行baseloop模块中的RunLoop接口，保证服务器持续进行新连接的获取+已建立连接上的通信
    */

    std::vector<LoopThread *> _threads;
    std::vector<EventLoop *> _eventloops;

public:
    LoopThreadPool(EventLoop *baseloop)
        : _thread_count(0), _next_loop_idx(0), _baseloop(baseloop)
    {
    }

    // 设置从属线程数量(从属Reactor)
    void SetThreadCount(int count)
    {
        _thread_count = count;
        _threads.resize(_thread_count);
        _eventloops.resize(_thread_count);
    }

    // 创建所有的从属LoopThread线程
    void CreateAllLoopThread()
    {
        if (_thread_count > 0)
        {
            for (int i = 0; i < _thread_count; i++)
            {
                _threads[i] = new LoopThread();
                _eventloops[i] = _threads[i]->GetEventLoop();
            }
        }
    }

    // 为通信连接分配EventLoop
    EventLoop *GetNextEventLoop()
    {
        // 1.若线程数量为0，则每次返回的都是baseloop。事件监控，就绪事件，任务队列等操作都在baseloop内完成
        if (_thread_count == 0)
            return _baseloop;
        // 2.若线程数量大于0，则每次采用RR轮转的思想，从_eventloops中循环遍历取出eventloop分配给Connection连接
        _next_loop_idx = (_next_loop_idx + 1) % _thread_count;
        return _eventloops[_next_loop_idx];
    }
};

// 保存任意自定义类型的应用层协议，应用层协议最终还是面向对象的，像该Http服务器中的HttpReqContext类型，还有可能是其他的自定义类型
class Any
{
    // 设计思想：在Any类里面搞一个多态，派生类用于保存任意类型的数据，而Any用于保存基类对象的指针，父类指针正好可以指向子类对象
private:
    class holder
    {
    public:
        holder()
        {
        }

        virtual ~holder()
        {
        }

        // 都是纯虚函数，迫使子类重写
        virtual const std::type_info &type() = 0;

        // 将来Any对象有可能进行拷贝构造，所以子类还能够克隆出一个新的对象
        virtual holder *clone() = 0;
    };

    template <class T>
    class placeholder : public holder
    {
    public:
        placeholder(const T &val)
            : _val(val)
        {
        }

        // 默认析构，对于内置类型不处理，对于自定义类型，调用他们类内的析构函数
        // 能执行析构函数进行资源的清理，则先执行析构函数。然后就调用operator delete完成对象所占用空间的释放销毁
        virtual ~placeholder()
        {
        }

        virtual const std::type_info &type()
        {
            return typeid(T);
        }

        virtual holder *clone()
        {
            return new placeholder(_val);
        }

    public:
        T _val; // 子类真正用于保存数据
    };

public:
    Any()
        : _content(nullptr)
    {
    }

    ~Any()
    {
        // 调用_data_ptr指向的placeholder对象内的析构函数，清理资源后调用free，释放_data_ptr变量所占用的空间
        if (_content)
            delete _content;
        _content = NULL;
    }

    // 模板函数，参数是任意类型，可以接收任意类型的数据
    // 通过特定类型数据推导 -> 创建Any对象
    template <class T>
    Any(const T &val)
        : _content(new placeholder<T>(val))
    {
    }

    // 通过其他Any对象创建一个新的Any对象
    Any(const Any &other)
        : _content(other._content == nullptr ? nullptr : other._content->clone())
    // 如果other中的父类指针不为空，则克隆一份新的数据对象出来，并将对象的地址拷贝给当前Any类对象保存的父类指针中，
    // 如果other中的父类指针为空，则也将当前Any对象中保存的父类指针置为NULL
    {
    }

    // 获取子类placeholder中保存的数据的指针
    template <class T>
    T *get()
    {
        // 用户想要获取的数据类型和Any中保存的数据对象的类型必须一致
        if (typeid(T) != _content->type())
            return nullptr;
        return &((placeholder<T> *)_content)->_val;
    }

    // 赋值运算符重载
    template <class T>
    Any &operator=(const T &val)
    {
        // 将构造出来的临时Any对象与当前的Any对象进行交换，则临时对象销毁时，原来的Any对象所保存的数据也就释放了
        Any tmp(val);
        std::swap(_content, tmp._content);
        return *this;
    }

    Any &operator=(Any other)
    {
        std::swap(_content, other._content);
        return *this;
    }

private:
    /*
    holder不是模板类，所以holder定义指针变量_data_ptr时，无须带有特定的类型，因为人家不是模板类！
    而需要是模板类来负责保存任意类型数据对象的是placeholder这个类，这样就可以完成Any的最主要的两个功能
        1.保存任意类型的数据(placeholder模板类来负责保存)
        2.构造Any对象时，无须带有特定的类型(->Any的成员变量不带有模板参数)
    */
    holder *_content;
};

class Connection;
using ConnPtr = std::shared_ptr<Connection>;

// 连接刚建立成功(各项回调函数还没被设置好) --> 通信状态(各项回调函数都被设置好了) -->
// 连接待关闭状态(等到处理完连接Buffer中滞留的数据后，状态转为关闭状态) --> 关闭状态
typedef enum
{
    INCOMPLETE,
    ESTABLISHED,
    TOBECLOSED,
    CLOSE
} ConnState;

// Connection模块是对Buffer模块，Socket模块，Channel模块的一个整体封装，实现了对一个通信套接字的整体的管理
// 每一个进行数据通信的套接字(也就是accept获取到的新连接)都会使用Connection进行管理
class Connection : public std::enable_shared_from_this<Connection>
{
private:
    // 连接的唯一ID，便于连接的管理和查找
    uint64_t _conn_id;
    // uint64_t _timer_id;              //定时器ID，必须是唯一的，这块为了简化操作使用_conn_id作为定时器ID

    int _sockfd;                   // 连接关联的文件描述符
    bool _enable_inactive_release; // 连接是否启动非活跃销毁的判断标志，默认为false
    ConnState _state;              // 连接的状态

    // 连接所关联的EventLoop(connection : Channel : TimerTask : EventLoop : LoopThread = 1 : 1 : 1 : 1 : 1)
    EventLoop *_loop;
    Socket _socket;     // 套接字操作管理
    Channel _chan;      // 连接的事件管理
    Buffer _in_buffer;  // 输入缓冲区 —— 存放从socket中读到的数据
    Buffer _out_buffer; // 输出缓冲区 ——  发送给对端的是数据，等到描述符事件可写，再发！
    Any _context;       // 请求的接受处理上下文

private:
    /*这四个回调函数，是让服务器模块来设置的(其实服务器模块的处理回调也是组件使用者设置的)*/
    /*换句话说，这几个回调都是组件使用者使用的*/
    using EstablishCallBack = std::function<void(const ConnPtr &)>;
    using MessageCallBack = std::function<void(const ConnPtr &, Buffer *)>;
    using CloseCallBack = std::function<void(const ConnPtr &)>;
    using AnyEventCallBack = std::function<void(const ConnPtr &)>;

    EstablishCallBack _establish_cb_usr;
    MessageCallBack _message_cb_usr;
    CloseCallBack _close_cb_usr;
    AnyEventCallBack _anyevent_cb_usr;

    // 组件内部设置的连接关闭回调，TcpServer模块内会把所有的连接管理起来，所以ConnPtr的最后一份引用技术会保存到TcpServer内的哈希表管理器中
    // 一旦某个连接要关闭就应该从管理的地方移除掉自己的信息
    CloseCallBack _close_cb_svr;

private:
    // 五个channel的事件回调函数

    // 描述符可读事件触发后调用的函数，接收socket数据放到接收缓冲区中，然后调用_message_callback
    void HandleRead()
    {
        // 1.接收sockfd的数据放到Buffer里
        char buf[65535]; // 64KB
        ssize_t ret = _socket.NonBlockRecv(buf, sizeof(buf) - 1);
        if (ret < 0)
        {
            // -1表示的是连接断开，即使出错了也不能直接关闭连接，需要看一下收发缓冲区中有没有数据滞留
            return ShutdownPriv();
        }

        // 这里的等于0表示的是没有读取到数据，而并不是连接断开了，连接断开返回的是-1
        _in_buffer.WriteAndMoveIdx(buf, ret);

        // 2.调用_message_cb进行业务逻辑处理
        if (_in_buffer.ReadableSize() > 0)
        {
            // shared_from_this从当前对象自身，获取自身的shared_ptr管理对象，当前对象所在类必须继承一个模板类enable_shared_from_this
            // 模板类内部会保存一份当前对象的weak_ptr智能指针，shared_from_this()接口其实就是通过weak_ptr来获取的shared_ptr对象
            return _message_cb_usr(shared_from_this(), &_in_buffer);
        }
    }

    void HandleWrite()
    {
        ssize_t ret = _socket.NonBlockSend(_out_buffer.ReadAddress(), _out_buffer.ReadableSize());
        if (ret < 0)
        {
            // 返回值<0，代表send出错，连接应该是断开了，则Release连接
            if (_in_buffer.ReadableSize() > 0)
                _message_cb_usr(shared_from_this(), &_in_buffer); // 如果输入缓冲区有数据，捎带处理一下

            return Release();
            /*
            将实际的释放连接操作压入到任务池中进行，因为在HandleReadyEvent中处理IO事件后还会刷新连接的活跃度，
            一旦HandleWrite函数中把连接真正释放了，那么刷新连接活跃度就会导致内存访问错误
            所以我们把连接的释放操作，放到就绪事件处理之后进行,也就是压入到任务池中进行
            */
        }

        // 发送成功，将输出缓冲区的位置偏移
        _out_buffer.MoveReadIdx(ret); // 千万不要忘了将读偏移位置向后偏移

        /*
        关闭写事件监控，因为写缓冲区中没有数据了，没啥可写的，自然就不需要关心sockfd上的写事件了
        否则会出现一种情况，sockfd上的写事件不断被触发，epoll_wait不断检测到写事件就绪，但_out_buffer中却一直没有数据发送，
        这其实就是在浪费epoll_wait的资源，所以在_out_buffer中没有数据之后，我们关闭其对于写事件的监控
        */
        if (_out_buffer.ReadableSize() == 0)
        {
            _chan.CancelWriteEvent();

            // 如果当前连接的状态是待关闭的状态，那在发送完_out_buffer中的内容之后，就应该释放连接，同时将连接状态置为关闭状态
            if (_state == TOBECLOSED)
                return Release();
        }
    }

    // HandleError()处理连接上发生的错误事件。当检测到错误时，此方法被调用
    // 由于错误通常意味着连接已不再可用，因此错误处理简单地通过调用 HandleClose()方法来进行，即将错误事件的处理转化为连接关闭的处理
    // 这样做可以确保在发生错误时，连接能够被适当关闭并释放资源
    void HandleError()
    {
        return HandleClose();
    }

    // 描述符连接关闭(对端关闭)
    void HandleClose()
    {
        // 一旦连接挂断了，套接字就什么都干不了了，这个挂断一般是客户端主动挂断连接，此时服务器即使_out_buffer中有数据，那也什么都做不了
        // 因为连接已经断开了。如果此时_in_buffer中还有数据的话，那我们服务器就捎带的进行一下数据的业务逻辑处理，然后再实际释放掉连接！
        if (_in_buffer.ReadableSize() > 0)
            _message_cb_usr(shared_from_this(), &_in_buffer);

        // 释放连接
        return Release();
    }

    // 设置描述符任意事件回调
    // 调用任意事件回调：如果设置了任意事件回调 (_event_callback)，则调用该回调函数
    // 这为组件使用者提供了一个机会，以便在发生任意事件时执行自定义逻辑。
    void HandleAnyEvent()
    {
        // 1. 刷新连接的活跃度 --- 延迟定时销毁任务
        if (_enable_inactive_release == true)
        {
            _loop->FlushTimer(_conn_id);
        }
        // 2. 调用组件使用者的任意事件回调
        if (_anyevent_cb_usr)
        {
            _anyevent_cb_usr(shared_from_this());
        }
    }

private:
    // 连接获取之后，所处的状态下要进行各种设置（启动读监控, 调用回调函数）
    void RefineConnPriv()
    {
        // 1.修改连接状态
        assert(_state == INCOMPLETE); // 此时连接各种回调函数的设置，以及读事件监控都未开启，所以连接状态一定是第一种(未完成的)
        _state = ESTABLISHED;
        // 2.启动读事件监控
        _chan.CareReadEvent();
        // 3.执行用户的回调函数
        if (_establish_cb_usr)
            _establish_cb_usr(shared_from_this());

        // 为什么不在构造函数中开启读事件监控呢？
        /*
        对于聊天室使用场景来讲，我们希望在连接刚建立完成后，给聊天室内的online user发送xxx上线的消息。如果在构造函数中开启读事件监控，
        则可能导致我们还没来得及调用_established_cb_usr给在线用户发送xxx上线的消息时，连接上已经收到了大量的消息了，我们希望的是，先
        发送xxx上线的消息，然后再进行连接上的通信

        所以理论上放到构造函数中也是可以的，只不过这有点不符合回调函数在连接建立后回调的规则，我们希望的是在连接正式建立完成之后再去
        进行_established_cb_usr的回调，所以我们将开启读事件的监控放到RefineConnectPriv中进行，开启读事件监控之后，
        立马进行_established_cb_usr的回调，这样就会先给在线用户广播xxx上线消息，然后再开始连接上的通信！
        */
    }

    // 实际释放连接的接口
    void ReleasePriv()
    {
        // 1. 修改连接状态，将其置为CLOSE
        _state = CLOSE;

        // 2. 移除连接的事件监控
        _chan.RemoveEvents();

        // 3. 关闭描述符
        _socket.Close();

        // 4. 如果当前定时器队列中还有定时销毁任务，则取消任务
        if (_loop->HasTimer(_conn_id))
            CloseInactPriv();

        // 5. 调用关闭回调函数，避免先移除服务器管理的连接信息导致Connection被释放，再去处理会出错，因此先调用用户的回调函数
        if (_close_cb_usr)
            _close_cb_usr(shared_from_this()); // 调用组件使用者设置的关闭回调函数

        // 6. 移除服务器内部管理的连接信息
        if (_close_cb_svr)
            _close_cb_svr(shared_from_this());
    }

    // 这个接口并不是实际的发送接口，而只是把数据放到了发送缓冲区，启动了可写事件监控
    void SendPriv(const Buffer &buf)
    {
        // 这个接口并不是真正的send数据，而是将要发送的数据放到_out_buffer里面，然后启动可写事件监控，等可写事件就绪后，就会回调HandleWrite
        // 在HandleWrite内部才会进行数据的实际发送。换句话说，触发可写事件之后，我们才进行数据的发送
        if (_state == CLOSE)
            return;

        _out_buffer.WriteAndMoveIdx(buf);
        if (_chan.IsWriteable() == false)
        {
            // 如果读事件的关心未开启,则开启对读事件的关心，读事件触发了，自动回调HandleWrite进行数据的发送
            _chan.CareWriteEvent();
        }
    }

    // 这个关闭操作并非实际的连接释放操作，还需要判断_in_buffer有没有数据待处理，_out_buffer待发送
    void ShutdownPriv()
    {
        // 修改连接为半关闭状态
        _state = TOBECLOSED;

        // 判断是输入缓冲区中是否有数据待处理
        if (_in_buffer.ReadableSize() > 0)
        {
            if (_message_cb_usr)
                _message_cb_usr(shared_from_this(), &_in_buffer);
        }

        // 判断输出缓冲区中是否有数据待发送
        if (_out_buffer.ReadableSize() > 0)
        {
            // 还有数据未发送，那么开启读事件的关心
            if (_chan.IsWriteable() == false)
                _chan.CareWriteEvent();
        }
        else if (_out_buffer.ReadableSize() == 0)
        {
            // 如果发送缓冲区中的数据已经发送完了，则将释放连接操作压入到任务队列中，等待全部就绪事件的业务处理完毕后再释放连接
            // Release();
            ReleasePriv();
        }
    }

    // 启动非活跃连接定时销毁，启动定时任务，定义多长时间无通信就是非活跃并销毁
    void OpenInactPriv(int seconds)
    {
        _enable_inactive_release = true;

        // 定时任务存在则刷新，不存在则添加
        if (_loop->HasTimer(_conn_id))
        {
            _loop->FlushTimer(_conn_id);
            return;
        }

        // 不能绑定那个直接释放连接的操作，而是要将释放连接的操作压入任务池，在就绪事件处理完成后，再去释放连接
        _loop->AddTimer(_conn_id, seconds, std::bind(&Connection::Release, this));
    }

    // 取消非活跃任务的销毁
    void CloseInactPriv()
    {
        _enable_inactive_release = false;
        if (_loop->HasTimer(_conn_id))
            _loop->CancelTimer(_conn_id);
    }

    // 切换协议的接口，重置上下文以及组件使用者的阶段性处理函数
    void SwitchProtocolPriv(const Any &context, const EstablishCallBack &establish_cb, const MessageCallBack &message_cb, const CloseCallBack &close_cb, const AnyEventCallBack &anyevent_cb)
    {
        _context = context;
        _establish_cb_usr = establish_cb;
        _message_cb_usr = message_cb;
        _close_cb_usr = close_cb;
        _anyevent_cb_usr = anyevent_cb;
    }

public:
    Connection(uint64_t conn_id, int sockfd, EventLoop *loop)
        : _conn_id(conn_id), _sockfd(sockfd), _loop(loop), _enable_inactive_release(false), _state(INCOMPLETE), _chan(_sockfd, _loop), _socket(sockfd)
    {
        // Channel模块各种回调函数的设置
        _chan.SetReadCallback(std::bind(&Connection::HandleRead, this));
        _chan.SetWriteCallback(std::bind(&Connection::HandleWrite, this));
        _chan.SetErrorCallback(std::bind(&Connection::HandleError, this));
        _chan.SetCloseCallback(std::bind(&Connection::HandleClose, this));
        _chan.SetAnyEventCallback(std::bind(&Connection::HandleAnyEvent, this));
    }

    ~Connection() { DLOG("release connection: %p !!!", this); }

    // 获取通信连接id
    uint64_t Id() { return _conn_id; }

    // 获取连接描述符
    int Fd() { return _sockfd; }

    // 判断通信连接是否已经建立完成，可以直接通信
    bool IsEstablished() { return _state == ESTABLISHED; }

    // 设置处理上下文
    void SetContext(const Any &context) { _context = context; }

    // 获取处理上下文
    Any *GetContext() { return &_context; }

    // 组件使用者设置给服务器模块，服务器模块获取 新的通信连接 后再设置给通信连接模块的回调函数
    void SetEstablishCbUsr(const EstablishCallBack &cb) { _establish_cb_usr = cb; }

    void SetMessageCbUsr(const MessageCallBack &cb) { _message_cb_usr = cb; }

    void SetCloseCbUsr(const CloseCallBack &cb) { _close_cb_usr = cb; }

    void SetAnyEventCbUsr(const AnyEventCallBack &cb) { _anyevent_cb_usr = cb; }

    // 服务器模块设置的连接关闭回调函数
    void SetCloseCbSvr(const CloseCallBack &cb) { _close_cb_svr = cb; }

    // 连接建立成功后进行各种状态设置 -- 启动可读事件监控、_establish_cb_usr()回调等
    void RefineConnect()
    {
        _loop->RunInLoop(std::bind(&Connection::RefineConnPriv, this));
    }

    // 提供给组件使用者的关闭接口 --- 并不实际关闭，需要判断有没有数据待处理
    void Shutdown()
    {
        _loop->RunInLoop(std::bind(&Connection::ShutdownPriv, this));
    }

    // 执行Release的线程一定是EventLoop线程自己，所以直接调用PushInTasks就行，当然你想要调用RunInLoop也可以，就是多执行了一个函数而已
    void Release()
    {
        _loop->PushInTasks(std::bind(&Connection::ReleasePriv, this));
        // _loop->RunInLoop(std::bind(&Connection::ReleasePriv, this));
    }

    // 启动非活跃销毁，并定义多长时间无通信就是非活跃，添加定时任务
    void OpenInactiveRelease(int seconds)
    {
        _loop->RunInLoop(std::bind(&Connection::OpenInactPriv, this, seconds));
    }

    // 取消非活跃销毁
    void CloseInactiveRelease()
    {
        _loop->RunInLoop(std::bind(&Connection::CloseInactPriv, this));
    }

    // Send，发送数据到缓冲区，启动写事件监控
    void Send(const char *data, size_t len)
    {
        // 外界传入的data，可能是个临时的空间，我们现在只是把发送操作压入了任务池，有可能并没有被立即执行
        // 因此有可能执行的时候，data指向的空间有可能已经被释放了，所以我们这里做一个数据的拷贝，拷贝到buf里面
        Buffer buf;
        buf.WriteAndMoveIdx(data, len);

        // 连接上接收到消息后，用户业务逻辑处理之后，可能要发送数据，但发送数据的实际操作只能在Eventloop线程内完成
        // 因为光一个连接内部所关联的channel在多线程环境下，都有可能产生线程安全，所以向外提供的这些函数最终都得在EventLoop的任务队列中执行
        _loop->RunInLoop(std::bind(&Connection::SendPriv, this, std::move(buf)));
        // RunInLoop只要一执行，就会唤醒阻塞的epoll_wait，让eventloop继续向下处理 就绪事件 + 执行任务池的任务
    }

    // 切换协议(切换数据的处理方式): 重置上下文和阶段性回调处理函数---这个接口必须在EventLoop的线程中立即执行，不可以压入到任务池执行
    void SwitchProtocol(const Any &context, const EstablishCallBack &establish_cb, const MessageCallBack &message_cb, const CloseCallBack &close_cb, const AnyEventCallBack &anyevent_cb)
    {
        /*
        存在一种情况，上层使用者调用该接口进行协议升级，但协议升级的这个API是放到任务池中进行的，很有可能这个API还没执行呢
        但此时连接上新就绪了事件，那么此时数据的处理还是会按照原来的协议方式进行处理，这样肯定是不行的！！！
        所以这个接口必须是EventLoop自己的线程来执行，外部的线程不能调用该接口，我们认为外部直接调用协议切换函数的这种方式是错误的！
        (目前该接口在服务器组件中没有用到)
        */
        _loop->AssertInLoop();
        _loop->RunInLoop(std::bind(&Connection::SwitchProtocolPriv, this, context, establish_cb, message_cb, close_cb, anyevent_cb));
    }
};

// Acceptor模块说实在的就是对监听套接字进行管理
// 我们这样就不用在服务端用listen_socket套接字了，而只需要用这一个Acceptor模块进行操作就可以了

// Accepter向外提供的接口其实只包含两个功能
// 一个是用于设置Accepter内部的_read_cb，也就是把分配Connection的函数设置成Accepter的回调接口
// 另一个是关心Accepter的可读事件，TcpServer的构造函数中会分别调用这两个接口

// Accepter模块的成员变量不应该包含Connection对象，所以为新连接fd分配Connection的工作是由TcpServer设置的，Accepter只负责回调而已
class Acceptor
{
    using ReadCallBack = std::function<void(int)>;

private:
    Socket _listen;   // 用于创建监听套接字
    EventLoop *_loop; // ⽤于对监听套接字进⾏事件监控
    Channel _chan;    // ⽤于对监听套接字进⾏事件管理

    // 回调函数，获取新连接时的回调函数，由服务器设置
    ReadCallBack _read_cb;

private:
    // 创建一个服务器连接并返回监听套接字
    int CreateServerConn(uint16_t port)
    {
        bool ret = _listen.CreateServerConn(port);
        assert(ret == true);
        return _listen.Fd();
    }

    // 监听套接字的读事件回调处理函数---获取新连接，调⽤_read_cb函数进⾏新连接处理
    // 获取新连接 + 为新连接分配一个Connection对象
    void HandleRead()
    {
        int newfd = _listen.Accpet();
        if (newfd < 0)
            return;
        if (_read_cb)
            _read_cb(newfd);
    }

public:
    Acceptor(EventLoop *loop, uint16_t port)
        : _listen(CreateServerConn(port)), _loop(loop), _chan(_listen.Fd(), loop)
    {
        _chan.SetReadCallback(std::bind(&Acceptor::HandleRead, this));
    }

    // 设置通信连接处理回调函数 --- 为获取到的新连接创建Connetion对象
    void SetReadCallBack(const ReadCallBack &cb) { _read_cb = cb; }

    // 启动读事件监控
    void CareReadEvent() { _chan.CareReadEvent(); }
};

// 上述我们就已经实现了高并发服务器的所有模块。这个模块就是将上述的所有模块进行了整合
// 通过 Tcpserver 模块实例化的对象，可以非常简单的完成一个服务器的搭建
/*
Tcpserver 模块主要管理的对象：
    Acceptor对象，创建一个监听套接字！
    EventLoop 对象，baseloop对象，实现对监听套接字的事件监控！
    std::vector conns,实现对新建连接的管理！
    LoopThreadPool 对象，创建loop线程池，对新建连接进行事件监控和处理！
*/

// 整合封装服务器模块
class TcpServer
{
private:
    // 通信连接的智能指针 -- 最开始的智能指针(引用计数为1)
    // using ConnectionPtr = std::shared_ptr<Connection>;

    // 组件使用者设置给服务器模块的各种事件回调
    using EstablishCallBack = std::function<void(const ConnPtr &)>;
    using MessageCallBack = std::function<void(const ConnPtr &, Buffer *)>;
    using CloseCallBack = std::function<void(const ConnPtr &)>;
    using AnyEventCallBack = std::function<void(const ConnPtr &)>;

    EstablishCallBack _establish_cb_usr;
    MessageCallBack _message_cb_usr;
    CloseCallBack _close_cb_usr;
    AnyEventCallBack _anyevent_cb_usr;

private:
    uint64_t _conn_id;             // 自增通信连接ID(也是定时器ID)
    uint16_t _port;                // 服务器监听的端口
    uint32_t _timeout;             // 定时任务的超时时间
    bool _enable_inactive_release; // 是否启动非活跃连接定时销毁任务的标志

    // 主EventLoop -- 主Reactor主线程，实现对监听套接字的事件监控
    EventLoop _baseloop;
    /*
        1.主线程创建该baseloop，该baseloop关联的线程id就是主线程的id，满足能够实现单Reactor服务器的demand
        其实这个baseloop就是主Reactor，负责获取可能产生的高并发连接请求，而其余的eventloop就是从属Reactor，负责通信连接上监控事件的处理，
        两者都是事件驱动处理模式
        2.如果eventloop数量为0，则baseloop自己负责lissock和sockfd两种套接字的三件套处理Monitor -> HandleReadyEvent -> ExecTasks
        baseloop自己调用RunLoop()，不断进行三件套的循环处理
        3.如果eventloop数量大于0，则baseloop负责lissock的三件套处理，eventloop负责sockfd的三件套处理，将两种套接字分到主从Reactor中
        进行事件监控，就绪事件处理，执行任务池任务
    */

    Acceptor _acceptor;                           // 监听套接字管理模块
    LoopThreadPool _pool;                         // 从属EventLoop线程池 -- 从Reactor从线程，实现对通信连接的事件监控
    std::unordered_map<uint64_t, ConnPtr> _conns; // 保存管理所有通信连接对象的shared_ptr指针

private:
    void AddTimerTaskPriv(const TaskFunc &task, int seconds)
    {
        // 这里把连接的id拿过来借用给定时任务
        _conn_id++;
        _baseloop.AddTimer(_conn_id, seconds, task); // 用户自己所添加的定时任务是在baseloop里面执行的
    }

    // 在EventLoop对应的线程中移除_conns中保存的Connection对象的shared_ptr
    void ReleaseConnPtrPriv(ConnPtr connptr)
    {
        int id = connptr->Id();
        auto it = _conns.find(id);

        // 销毁最后一份哈希表中存储的conn连接对象，实现连接的真正释放
        if (it != _conns.end())
        {
            _conns.erase(it);
        }
    }

    // 通信连接关闭时移除_conns中保存的最后一份Connection对象的shared_ptr，从而释放掉连接对应的Connection对象
    void ReleaseConnPtr(ConnPtr connptr)
    {
        // 服务器模块的通信连接的彻底销毁，最终其实是在baseloop里面执行的，其他从属线程调用的_close_cb_svr，其实是将释放连接的操作，
        // 压入到了baseloop的任务队列中执行的，压入到任务池，并不是就泡在里面迟迟不执行了，相反很快就会执行，因为只要压入任务池，就会立马
        // 唤醒可能阻塞的epoll_wait，向下进行就绪事件处理，然后执行任务池的任务
        _baseloop.RunInLoop(std::bind(&TcpServer::ReleaseConnPtrPriv, this, connptr));
    }

    // 为获取到的新连接创建Connection对象并管理其shared_ptr
    void NewConnection(int sockfd)
    {
        _conn_id++;
        // Connection的构造函数会设置好sockfd的五个事件回调函数
        ConnPtr conn(new Connection(_conn_id, sockfd, _pool.GetNextEventLoop()));

        // 设置组件使用者的五个回调函数
        conn->SetEstablishCbUsr(_establish_cb_usr);
        conn->SetMessageCbUsr(_message_cb_usr);
        conn->SetCloseCbUsr(_close_cb_usr);
        conn->SetAnyEventCbUsr(_anyevent_cb_usr);
        conn->SetCloseCbSvr(std::bind(&TcpServer::ReleaseConnPtr, this, std::placeholders::_1));

        // 添加连接的定时器任务
        if (_enable_inactive_release == true)
            conn->OpenInactiveRelease(_timeout);

        // 上面连接的构造函数初始化好后，再开启通信套接字sockfd的读事件监控
        conn->RefineConnect();

        /* <定时任务的添加, 开启读事件监控> 这两个操作之间有时序问题，必须先进行定时任务的添加，然后再开启读事件的监控

        ---反过来错误的原因:
        先开启读事件监控时，可能立马有事件到来，这时候anyeventCallback会进行定时任务的刷新，但这个时候时间轮里面还没有定时任务呢，
        因为是后进行的定时任务添加，所以反过来是不可行的！

        ---不会产生时序问题的原因：
        开启定时任务销毁的这个函数会被压入到任务池中，而PushInTasks的操作会唤醒epoll_wait，主线程运行完PushInTasks后，回到1814行
        继续向下开启读事件监控，但在主线程运行完PushInTasks的一瞬间，EventLoop线程就会HandleReadyEvent了，下一步就执行任务池中的任务，
        而当主线程开启读事件监控时，EventLoop线程就算是最慢最慢，他也会进行到HandleReadyEvent，而处理的事件中一定是不包含当前连接的，
        所以一定不会刷新当前连接相关的定时任务，因为EventLoop线程早就epoll_wait结束了，即使当前我们的连接上有读事件就绪，那也一定是
        EventLoop线程执行完任务池中的任务，而任务中就包含了添加当前连接对应的定时销毁任务，所以当前连接读事件就绪时，定时任务早就被
        添加到时间轮里面了，此时刷新定时任务就不会出错了。*/

        _conns.insert(std::make_pair(_conn_id, conn));
    }

public:
    TcpServer(uint16_t port)
        : _port(port), _conn_id(0), _enable_inactive_release(false), _acceptor(&_baseloop, port), _pool(&_baseloop)
    {
        // 设置acceptor的回调函数
        _acceptor.SetReadCallBack(std::bind(&TcpServer::NewConnection, this, std::placeholders::_1));

        // 关心acceptor的读事件
        _acceptor.CareReadEvent();
        DLOG("TcpServer构造完毕!!!");
    }

    // 组件使用者设置给服务器模块的各种事件回调
    void SetEstablishCbUsr(const EstablishCallBack &cb) { _establish_cb_usr = cb; }

    void SetMessageCbUsr(const MessageCallBack &cb) { _message_cb_usr = cb; }

    void SetCloseCbUsr(const CloseCallBack &cb) { _close_cb_usr = cb; }

    void SetAnyEventCbUsr(const AnyEventCallBack &cb) { _anyevent_cb_usr = cb; }

    // 设置从属线程(子Reactor)数量
    void SetThreadCount(int count) { _pool.SetThreadCount(count); }

    // 启动服务器
    void Run()
    {
        // 服务器模块的运行，具象化描述就是事件驱动处理模式 -> eventloop.runloop(); 三件套:Monitor -> HandleReadyEvent -> ExecTasks
        // baseloop和eventloop都是最终调用RunLoop运行，不断执行三件套
        _pool.CreateAllLoopThread();
        _baseloop.RunLoop();
    }

    // 启动非活跃释放连接
    void OpenInactiveRelease(int timout)
    {
        _enable_inactive_release = true;
        _timeout = timout;
    }

    // 用于添加用户自身的定时任务的接口
    void AddTimerTask(const TaskFunc &task, int seconds)
    {
        _baseloop.RunInLoop(std::bind(&TcpServer::AddTimerTaskPriv, this, task, seconds));
    }
};

void Channel::RemoveEvents() { return _loop->RemoveEvent(this); }
void Channel::UpdateEvents() { return _loop->UpdateEvent(this); }

void TimerQueue::AddTimer(uint64_t tid, int timeout, TaskFunc task_cb) { _loop->RunInLoop(std::bind(&TimerQueue::AddTimerPriv, this, tid, timeout, task_cb)); }
void TimerQueue::FlushTimer(uint64_t tid) { _loop->RunInLoop(std::bind(&TimerQueue::FlushTimerPriv, this, tid)); }
void TimerQueue::CancelTimer(uint64_t tid) { _loop->RunInLoop(std::bind(&TimerQueue::CancelTimerPriv, this, tid)); }

class Network
{
public:
    Network()
    {
        DLOG("SIGPIPE IGNORE INIT!!!");
        signal(SIGPIPE, SIG_IGN);
        // 当连接断开后，如果服务器还继续向连接发送数据，此时就会触发异常SIGPIPE，操作系统就会给我们的服务器进程发送SIGPIPE，
        // 进程就会终止退出，所以我们让服务器忽略掉SIGPIPE信号
    }
};

static Network nw;

#endif